package com.pw.study.realtime.handle

import com.pw.study.realtime.utils.MyUtils
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}

import java.util.Properties

object KafkaHandler {

  private val properties: Properties = MyUtils.load("config.properties")
  private val broker_list: String = properties.getProperty("kafka.broker.list")
  private var commitBtn: String = properties.getProperty("kafka.offset.enable.auto.commit")
  def getKafkaStream(topics: Array[String], context: StreamingContext, groupName: String): InputDStream[ConsumerRecord[String, String]] = {
    if(commitBtn.equals("true")){
      commitBtn="false"
    }
    context.sparkContext.setLogLevel("error")
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> broker_list,
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> groupName,
      "auto.offset.reset" -> "earliest",
      "enable.auto.commit" -> commitBtn
    )
    val stream = KafkaUtils.createDirectStream(context, LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams))
    stream
  }
  def getKafkaStreamByAutoCommit(topics: Array[String], context: StreamingContext, groupName: String): InputDStream[ConsumerRecord[String, String]] = {
    if(commitBtn.equals("false")){
      commitBtn="true"
    }
    val stream = getKafkaStream(topics, context, groupName)
    stream
  }
  def getKafkaStreamByMysql(topics: Array[String], context: StreamingContext, groupName: String,offsetsMap: Map[TopicPartition, Long]): InputDStream[ConsumerRecord[String, String]] = {
    if(commitBtn.equals("true")){
      commitBtn="false"
    }
    context.sparkContext.setLogLevel("error")
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> broker_list,
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> groupName,
      "auto.offset.reset" -> "earliest",
      "enable.auto.commit" -> commitBtn
    )
    val stream = KafkaUtils.createDirectStream(context, LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams,offsetsMap))
    stream
  }
}
